package com.jianyitech.asr.kafka.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.jianyitech.asr.dtos.Tenant;
import com.jianyitech.asr.dtos.TenantIp;
import com.jianyitech.asr.kafka.producer.ProducerService;
import com.jianyitech.asr.service.TenantService;
import com.jianyitech.asr.utils.Constant;
import com.jianyitech.asr.utils.StringUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.cache.CacheProperties;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
public class TenantAccountConsumer {

    private static final Logger logger = LoggerFactory.getLogger(TenantAccountConsumer.class);


    @Autowired
    TenantService tenantService;

    @Autowired
    ProducerService producerService;
    @Autowired
    StringRedisTemplate redisTemplate;

    @KafkaListener(topics="tp_asr_heartbeat",groupId = "asr-client")
    public void onTenantConsumer(List<ConsumerRecord<String,String>> msgs){
        if(null!=msgs && !msgs.isEmpty()){
            for(ConsumerRecord<String,String> record : msgs){
                String value = record.value();
                if(StringUtil.isPresent(value)){
                    TenantIp tenantIp = JSONObject.parseObject(value,TenantIp.class);
                    Tenant tenant = tenantService.findById(tenantIp.getTenantId());
                    if(null!=tenant){
                        loadTenantCache(tenantIp);
                        producerService.sendMessage("tp_asr_account_"+tenant.getId()+"_"+tenantIp.getIp(), JSON.toJSONString(tenant));

                    }

                }
            }
        }
    }

    @KafkaListener(topics="tp_asr_vos_record",groupId = "asr-client")
    public void tenantConsumptionConsumer(List<ConsumerRecord<String,String>> msgs){
        if(null!=msgs && !msgs.isEmpty()){
            for(ConsumerRecord<String,String> record : msgs){
                String value = record.value();
                if(StringUtil.isPresent(value)){
                    redisTemplate.opsForList().rightPush("VOS_RECORD",value);
                }
            }
        }
    }


    /**
     * 加载tenant-ip的cache
     * @param tenantIp
     */
    private void loadTenantCache(TenantIp tenantIp){
        redisTemplate.opsForHash().put(Constant.CACHE.TENANT_HEARTBEAT_CACHE_PREFIX+tenantIp.getTenantId(),tenantIp.getIp(),tenantIp.getTs()+"");
    }

}
